package com.pactera.elasticsearch.autoconfigure;

import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.elasticsearch.client.Node;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Arrays;
import java.util.Objects;

/**
 * @author: liqingdian
 **/
@Configuration
public class ElasticsearchConfiguration {

    private static final Logger log = LoggerFactory.getLogger(ElasticsearchConfiguration.class);

    @Autowired
    private ElasticsearchProperties properties;

    @Bean
    @ConditionalOnMissingBean
    public RestClientBuilder restClientBuilder() {
        // 如果有多个从节点可以持续在内部new多个HttpHost，参数1是ip,参数2是HTTP端口，参数3是通信协议
        HttpHost[] httpHost = createHttpHost();
        RestClientBuilder clientBuilder = RestClient.builder(httpHost);
        // 设置线程数
        setHttpClientConfigCallback(clientBuilder);
        // 设置超时时间
        setRequestConfigCallback(clientBuilder);
        // 失败监听
        setFailureListener(clientBuilder);
        // 最后配置好的clientBuilder再build一下即可得到真正的Client
        return clientBuilder;
    }

    @Bean(destroyMethod = "close")
    @ConditionalOnBean(name = "restClientBuilder")
    public RestHighLevelClient restHighLevelClient() {
        return new RestHighLevelClient(restClientBuilder());
    }

    /**
     * 连接配置
     */
    private HttpHost[] createHttpHost() {
        HttpHost[] hosts = Arrays.stream(properties.getHostnames().split(","))
                .map(this::makeHttpHost)
                .filter(Objects::nonNull)
                .toArray(HttpHost[]::new);
        return hosts;
    }

    /**
     * 创建HttpHost
     */
    private HttpHost makeHttpHost(String hostname) {
        assert StringUtils.isNotBlank(hostname);
        if (hostname.indexOf("://") > -1) {
            return HttpHost.create(hostname);
        }
        if (hostname.lastIndexOf(":") > 6) {
            String[] address = hostname.split(":");
            String ip = address[0];
            int port = Integer.parseInt(address[1]);
            return new HttpHost(ip, port, properties.getHttpScheme());
        }
        return new HttpHost(hostname, 9200, properties.getHttpScheme());
    }

    /**
     * 默认配置
     */
    private void setHttpClientConfigCallback(RestClientBuilder clientBuilder) {
        CredentialsProvider credentialsProvider = null;
        if (properties.isSecurityEnabled()) {
            // 设置密码
            credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(properties.getUsername(), properties.getPassword()));
        }
        final CredentialsProvider cp = credentialsProvider;
        /* 配置异步请求的线程数量，Apache Http Async Client默认启动一个调度程序线程，以及由连接管理器使用的许多工作线程
        （与本地检测到的处理器数量一样多，取决于Runtime.getRuntime().availableProcessors()返回的数量）。线程数可以修改如下,
        这里是修改为1个线程，即默认情况
        */
        clientBuilder.setHttpClientConfigCallback(httpAsyncClientBuilder -> {
            if (cp != null) {
                httpAsyncClientBuilder.setDefaultCredentialsProvider(cp);
            }
            httpAsyncClientBuilder.setMaxConnTotal(properties.getMaxConnTotal());
            httpAsyncClientBuilder.setMaxConnPerRoute(properties.getMaxConnPerRoute());
            httpAsyncClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(properties.getThreadCount()).build());
            return httpAsyncClientBuilder;
        });
    }

    /**
     * 默认回调
     */
    private void setRequestConfigCallback(RestClientBuilder clientBuilder) {
        clientBuilder.setRequestConfigCallback(requestConfigBuilder -> {
            // 连接5秒超时，套接字连接60s超时
            requestConfigBuilder.setConnectTimeout(properties.getConnectTimeout());
            requestConfigBuilder.setSocketTimeout(properties.getSocketTimeout());
            requestConfigBuilder.setConnectionRequestTimeout(properties.getConnectionRequestTimeout());
            return requestConfigBuilder;
        });
    }

    /**
     * 失败监听
     */
    private void setFailureListener(RestClientBuilder clientBuilder) {
        clientBuilder.setFailureListener(new RestClient.FailureListener() {
            @Override
            public void onFailure(Node node) {
                super.onFailure(node);
                log.error("\nElasticsearch异常: {}", node.getName());
            }
        });
    }
}
